package org.artifact.core.plugin.rocketmq;

import cn.hutool.log.Log;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class RocketMQConsumer {
    private String namesrvAddr = null;
    private String groupName = null;
    private String consumerTopic = null;
    private String consumerTag = null;
    private DefaultMQPushConsumer consumer;
    public void start(){
        try{
            consumer = new DefaultMQPushConsumer(groupName);

            //同样也要设置NameServer地址
            consumer.setNamesrvAddr(namesrvAddr);

            //这里设置的是一个consumer的消费策略
            //CONSUME_FROM_LAST_OFFSET 默认策略，从该队列最尾开始消费，即跳过历史消息
            //CONSUME_FROM_FIRST_OFFSET 从队列最开始开始消费，即历史消息（还储存在broker的）全部消费一遍
            //CONSUME_FROM_TIMESTAMP 从某个时间点开始消费，和setConsumeTimestamp()配合使用，默认是半个小时以前
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

            //设置consumer所订阅的Topic和Tag，*代表全部的Tag
            consumer.subscribe(consumerTopic, consumerTag);


//            consumer.
            //设置一个Listener，主要进行消息的逻辑处理
            consumer.registerMessageListener(new MessageListenerOrderly() {
                @Override
                public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
                    consumeOrderlyContext.setAutoCommit(true);
                    for (MessageExt messageExt : list) {
                        System.out.println(messageExt.getUserProperty("orderId") + " " + new String(messageExt.getBody()));
                    }
                    return ConsumeOrderlyStatus.SUCCESS;
                }
            });

            //调用start()方法启动consumer
            consumer.start();
            Log.get().info("启动成功");
        }catch (Exception e){

        }

    }
    public void stop(){
        consumer.shutdown();
    }

    public static void main(String[] args) {
        RocketMQConsumer consumer = new RocketMQConsumer();
        consumer.consumerTag = "*";
        consumer.namesrvAddr = "localhost:9876";
        consumer.consumerTopic = "Demo";
        consumer.groupName = "defaultGroup";

        consumer.start();
    }
}
